package com.suxin.kafka.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

/**
 * @Classname KafkaAdmin
 * @Description [ KafkaAdmin ]
 * @Author Tang
 * @Date 2020/11/14 23:40
 * @Created by ASUS
 */
@Configuration
public class KafkaAdminConfig {

    @Bean
    public NewTopic oneTopic() {
        return TopicBuilder.name("one-kafka-test")
                .partitions(1)
                .replicas(0)
                .compact()
                .build();
    }

    @Bean
    public NewTopic twoTopic() {
        return TopicBuilder.name("two-kafka-test")
                .partitions(1)
                .replicas(0)
                .compact()
                .build();
    }

    @Bean
    public NewTopic threeTopic() {
        return TopicBuilder.name("three-kafka-test")
                .partitions(2)
                .replicas(0)
                .compact()
                .build();
    }


    @Bean
    public NewTopic batchTopic() {
        return TopicBuilder.name("batch-kafka-test")
                .partitions(1)
                .replicas(0)
                .compact()
                .build();
    }


    @Bean
    public NewTopic fourTopic() {
        return TopicBuilder.name("four-kafka-test")
                .partitions(5)
                .replicas(0)
                .compact()
                .build();
    }

    @Bean
    public NewTopic fiveTopic() {
        return TopicBuilder.name("five-kafka-test")
                .partitions(5)
                .replicas(0)
                .compact()
                .build();
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(
            @Qualifier("producerConfigs") KafkaProperties producerConfigs) {
        return new KafkaTemplate<>(producerFactory(producerConfigs));
    }

    @Bean
    public ProducerFactory<String, String> producerFactory(
            @Qualifier("producerConfigs") KafkaProperties producerConfigs) {
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(
                producerConfigs.buildProducerProperties()
        );
        // 会创建 线程私有 KafkaProducer ，默认创建出来是单例的
        producerFactory.setProducerPerThread(true);
        return producerFactory;
    }

    @Primary
    @Bean
    @ConfigurationProperties(prefix = "spring.kafka")
    public KafkaProperties producerConfigs() {
        return new KafkaProperties();
    }

}